package streamAPI.source;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.NumberSequenceIterator;
import org.testng.annotations.Test;

/** 一些简单Source 的使用合集
 * @author yue.cao
 * @since 10-29-2020
 */
public class DemoWithEasySource {

	@Test
	public void demoWithMyNoParallelSource() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Long> source = env.addSource(new MyNoParallelSource());
		DataStream<String> map = source.map(new MapFunction<Long, String>() {
			@Override
			public String map(Long in) throws Exception {
				return "===>" + in;
			}
		});
		map.print();
		env.execute();
	}

	@Test
	public void demoWithMyParallelSource() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Long> source = env.addSource(new MyParallelSource()).setParallelism(2);
		DataStream<String> map = source.map(new MapFunction<Long, String>() {
			@Override
			public String map(Long in) throws Exception {
				return "===>" + in;
			}
		});
		map.print();
		env.execute();
	}

	@Test
	public void demoWithMyRichParallelSource() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Long> source = env.addSource(new MyRichParallelSource()).setParallelism(2);
		DataStream<String> map = source.map(new MapFunction<Long, String>() {
			@Override
			public String map(Long in) throws Exception {
				return "===>" + in;
			}
		});
		map.print();
		env.execute();
	}

	@Test
	public void demoWithWanJuSource() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		//带并行度的玩具source
		DataStreamSource<Long> source = env.fromParallelCollection(new NumberSequenceIterator(0, 9), Long.class);
		System.out.println("=========>" + source.getParallelism());
		source.filter(new FilterFunction<Long>() {
			@Override
			public boolean filter(Long value) throws Exception {
				return value % 2 == 0;
			}
		}).print();
		env.execute();
	}




}
